package hit.edu.zjc.Operator;

import hit.edu.zjc.DataClean.DataCleaner;
import hit.edu.zjc.DataIntegration.DataInteDis;
import hit.edu.zjc.Operator.Entity.DataIDSimilarityPair;
import hit.edu.zjc.Tool.ObjectConvert;
import hit.edu.zjc.Tool.RWObject;
import hit.edu.zjc.Tool.ShowOutPut;
import hit.edu.zjc.UserInterface.UDColumInfo;
import hit.edu.zjc.UserInterface.UserDef;
import hit.edu.zjc.marshalling.ByteSerializerDeserializer;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.File;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Random;

import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.Integer64SerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;

public class SimilarityComputeOperatorDescriptor extends
		AbstractSingleActivityOperatorDescriptor {

	private static final long serialVersionUID = 1L;
	private ISerializerDeserializer[] DataISD_array = null;
	private UDColumInfo[] Udci_array = null;
	private String FilePath = null;//GramMapCompleted file path
	private Hashtable<String, HashSet<Integer>>[] GramMap = null;
	private Hashtable<Integer,HashSet<Integer>> SetTable=new Hashtable<Integer,HashSet<Integer>>();
	private ArrayList<HashSet<Integer>> SetList=new ArrayList<HashSet<Integer>>();

	public SimilarityComputeOperatorDescriptor(JobSpecification spec,
			ISerializerDeserializer[] isd_array, UDColumInfo[] udci_array,
			String filepath) {
		super(spec, 1, 2);// 输入输出端的数量。0输出SetList，1输出SetTable
		DataISD_array = isd_array;
		Udci_array = udci_array;
		recordDescriptors[0] = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});// 输出数据的格式。
		recordDescriptors[1] = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});
		FilePath = filepath;

		/*
		 * 不许要按顺序 //检查udci_array是否按ColumID递增顺序排列。 for(int
		 * i=1;i<Udci_array.length;i++) { for(int j=i;j>0;j--) {
		 * if(Udci_array[j].ColumID>Udci_array[j-1].ColumID) break; else {
		 * UDColumInfo udci=Udci_array[j]; Udci_array[j]=Udci_array[j-1];
		 * Udci_array[j-1]=udci; } } }
		 */
	}

	@Override
    public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {

        final FrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), new RecordDescriptor(DataISD_array));

        return new AbstractUnaryInputOperatorNodePushable() {

        	  private final IFrameWriter[] writers = new IFrameWriter[outputArity];
        	  
            private ByteBuffer[] outputBuffer=new ByteBuffer[outputArity];

            
            private void showSet(HashSet<Integer>set)
            	{
            		System.out.println("Set Size:"+set.size());
            		Iterator<Integer>it=set.iterator();
            		while(it.hasNext())
            		{
            			Integer id=it.next();
            			System.out.print(id+" ,");
            		}
            	
            	}
            
            private void showArrayList(ArrayList<HashSet<Integer>> SetList)
               {
                 for(int i=0;i<SetList.size();i++)
                   {
              	   HashSet<Integer>set=SetList.get(i);
              	   System.out.println("Set: "+(i+1)+"th");
              	   Iterator<Integer> iterator=set.iterator();
              	   while(iterator.hasNext())
              	   {
              		   Integer id=iterator.next();
              		   System.out.print(""+id+", ");
              	   }
              	   System.out.println();
              	   
                   }
                 System.out.println("SetList size:"+SetList.size());
               }
            
              //判断连个集合是否相交
            public boolean isCover(HashSet<Integer>set1,HashSet<Integer>set2)
               {
            	   Iterator<Integer>iterator=set1.iterator();
          	   while(iterator.hasNext())
          	   		{
          		   Integer data=iterator.next();
          		   boolean res=set2.contains(data);
          		   if(res)
          			   return true;
          	   		}
          	   return false;
               }

            @Override
            public void open() throws HyracksDataException {
                outputBuffer[0] = ctx.allocateFrame();
                outputBuffer[1] = ctx.allocateFrame();
                GramMap=(Hashtable<String,HashSet<Integer>>[])RWObject.ReadObject(FilePath);//读取数据。
                for(int i=0;i<outputArity;i++)
                	writers[i].open();
            }

            @Override
            public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
                accessor.reset(buffer);
                int tupleCount = accessor.getTupleCount();
                int fieldCount = accessor.getFieldCount();
                byte[] byte_array=accessor.getBuffer().array();
                int fieldSlotsLength=accessor.getFieldSlotsLength();
                for (int i = 0; i < tupleCount; i++) {
                    int tupleStartOffset = accessor.getTupleStartOffset(i);//获得第i个Tuple的开始位置。
                    int tupleEndOffset=accessor.getTupleEndOffset(i);
                    try {
                    	
                        int udci_i=0;//udci_array的index。
                        int Data_ID=0;//数据条目的主键值。

                              //获取数据条目DataID的值：
							  int fieldstartoffset=accessor.getFieldStartOffset(i, 0);
							  int fieldendoffset=accessor.getFieldEndOffset(i, 0);
							  int fieldlength=accessor.getFieldLength(i, 0);   
							  String data_str=UTF8StringSerializerDeserializer.INSTANCE.deserialize( 
									new DataInputStream(
											new ByteArrayInputStream(byte_array,tupleStartOffset+fieldSlotsLength+fieldstartoffset,fieldlength)));
							  Data_ID=Integer.parseInt(data_str);//记录数据条目的主键值。
							  
							  DataIDSimilarityPair pair=new DataIDSimilarityPair();
							  for(int k=0;k<Udci_array.length;k++)
							   {
								  int fIdx=Udci_array[k].ColumID;						   
							     fieldstartoffset=accessor.getFieldStartOffset(i, fIdx);
		                     fieldendoffset=accessor.getFieldEndOffset(i, fIdx);
		                     fieldlength=accessor.getFieldLength(i, fIdx);   
		                     
		                     data_str=UTF8StringSerializerDeserializer.INSTANCE.deserialize( 
		                    			new DataInputStream(
		                    					new ByteArrayInputStream(byte_array,tupleStartOffset+fieldSlotsLength+fieldstartoffset,fieldlength)));
		                     data_str=data_str.toUpperCase();
		                     data_str=data_str.replaceAll(" ", "");
		                          //开始计算相似度：计算1Gram的相似度
		                     for(int str_index=0;str_index<data_str.length();str_index++)
		                     	  {
		                    	   String substring=data_str.substring(str_index, str_index+1);//取出1Gram
		                    	 	HashSet<Integer>set=GramMap[k].get(substring);
		                    	 	if(set!=null)
		                    	 		{
		                    	 		Iterator<Integer> iterator=set.iterator();
		                    	 		while(iterator.hasNext())
		                    	 			{
		                    	 			Integer data_id=iterator.next();
		                    	 			double weight=1.0/data_str.length();
		                    	 			pair.addOneGramWeight(data_id, weight);
		                    	 			}
		                    	 		}
		                          }
		                     	//计算2Gram相似度。
		                    for(int str_index=0;str_index<data_str.length()-1;str_index++)
		                     	 {
		                    	 String substring=data_str.substring(str_index, str_index+2);
		                    	 HashSet<Integer>set=GramMap[k].get(substring);
		                    	 if(set!=null)
		                    	 	 {
		                    		 Iterator<Integer>iterator=set.iterator();
		                    		 while(iterator.hasNext())
		                    		 	 {
		                    			 Integer data_id=iterator.next();
		                    			 double weight=1.0/(data_str.length()-1);
		                    			 pair.addTwoGramWeight(data_id,weight);
		                    		 	 }
		                    	 	 }
		                         }
		                    pair.refresh(Udci_array[k].Weight);//找出最大的weight
						       }
							 HashSet<Integer> set=pair.getKeySet(UDColumInfo.GetThreshold());//相似的set。
							 HashSet<Integer> inputset=new HashSet<Integer>();
							 set.add(Data_ID);//set中应该有了，加这一句，以防万一：threshold大于所有列权值的和
							 inputset.addAll(set);//复制一份放在hashtable内。
							 if(inputset.size()>1)
								 SetTable.put(Data_ID, inputset);
							 
							 //ShowOutPut.show("set size:"+set.size());
			 
							 int k=0;
							 ArrayList<Integer> CoverIndex=new ArrayList<Integer>();
							 for(k=0;k<SetList.size();k++)
                       		 {
                    	   HashSet<Integer> otherset=SetList.get(k);
                    	   boolean iscover=isCover(set,otherset);
                    	   if(iscover)//相交
                    	   		{
                    		   	CoverIndex.add(0,k);//构成降序列
                    	   		}
                       		 }
                       if(CoverIndex.size()==0)//没有相交的集合。
                       		 {
                    	   SetList.add(set);
                       		 }
                       else//处理相交的情况。
                       		{
                   			for(int m=0;m<CoverIndex.size();m++)
                   				{
                   				int index=CoverIndex.get(m);
                   				set.addAll(SetList.get(index));
                   				SetList.remove(index);
                   				}
                   			SetList.add(set);
                       		}
                         	
							 
                    	} 
                    catch (Exception e) {
								e.printStackTrace();
								//System.out.println("产生异常,原因是字符串编码不是UTF-8.");
						 }
                	}
                //showArrayList(SetList);
            	}

            @Override
            public void fail() throws HyracksDataException {
            	
            		for (IFrameWriter writer : writers) 
            		{
                    writer.fail();
                	}
               }

            
            private void WriteByteBuffer(int index,byte[] data) throws HyracksDataException
            	{
            		outputBuffer[index].clear();
            	   if(data.length<outputBuffer[index].capacity())
               		{
                   outputBuffer[index].put(data);
                   FrameUtils.flushFrame(outputBuffer[index], writers[index]);
               		}
               else//一个frame装不下
               		{
               		int byteremain=data.length;
               		int i=0;
               		for(i=0;byteremain>=outputBuffer[index].capacity();i++)
               			{
               			outputBuffer[index].put(data,i*outputBuffer[index].capacity(),outputBuffer[index].capacity());
                   		FrameUtils.flushFrame(outputBuffer[index], writers[index]);
                   		outputBuffer[index].clear();
                   		byteremain=byteremain-outputBuffer[index].capacity();
               			}
               		if(byteremain>0)
               			{
               			outputBuffer[index].put(data,i*outputBuffer[index].capacity(),byteremain);
               			FrameUtils.flushFrame(outputBuffer[index], writers[index]);
               			}
               		}
            	}
            
            @Override
            public void close() throws HyracksDataException {
            		//?
                byte[] objbytes=ObjectConvert.ObjectToByteArray(SetList);
                WriteByteBuffer(0,objbytes);
                objbytes=ObjectConvert.ObjectToByteArray(SetTable);
                WriteByteBuffer(1,objbytes);
                outputBuffer[0]=null;
                outputBuffer[1]= null;
                for(IFrameWriter writer:writers)
                	writer.close();
            }

			@Override
			public void setOutputFrameWriter(int index, IFrameWriter writer,
					RecordDescriptor rDesc)
			{
				 writers[index] = writer;
				
			}
        };
    }
}